package redis.search;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.redis.bolt.RedisLookupBolt;
import org.apache.storm.redis.common.config.JedisPoolConfig;
import org.apache.storm.redis.common.mapper.RedisDataTypeDescription;
import org.apache.storm.redis.common.mapper.RedisLookupMapper;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.ITuple;
import org.apache.storm.tuple.Values;
import redis.WordSpout;

import java.util.ArrayList;
import java.util.List;

import static redis.PersistentWordCount.HASH_KEY_NAME;

public class LookupWordCount {

    /**
     * 从Redis中取数据并发射
     */
    private static class WordCountLookupMapper implements RedisLookupMapper {
        private final RedisDataTypeDescription description;

        public WordCountLookupMapper() {
            description = new RedisDataTypeDescription(RedisDataTypeDescription.RedisDataType.HASH, HASH_KEY_NAME);
        }

        @Override
        public RedisDataTypeDescription getDataTypeDescription() {
            return description;
        }

        /**
         * 将查询结果组装为Storm的Values，并向下游发射
         * @param iTuple 上游的Tuple
         * @param o 查询的结果
         * @return 发到下游的Tuple
         */
        @Override
        public List<Values> toTuple(ITuple iTuple, Object o) {
            String keyFromTuple = getKeyFromTuple(iTuple);

            List<Values> values = new ArrayList<>();
            values.add(new Values(keyFromTuple, o)); // o is result by iTuple key

            return values;
        }

        @Override
        public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
            outputFieldsDeclarer.declare(new Fields("word", "count"));
        }

        @Override
        public String getKeyFromTuple(ITuple iTuple) {
            return iTuple.getStringByField("word");
        }

        @Override
        public String getValueFromTuple(ITuple iTuple) {
            return null;
        }
    }

    public static void main(String[] args) throws Exception {
        JedisPoolConfig poolConfig = new JedisPoolConfig.Builder().setHost("liu").setPort(6379).build();

        WordSpout wordSpout = new WordSpout();
        WordCountLookupMapper wordCountLookupMapper = new WordCountLookupMapper();
        RedisLookupBolt redisLookupBolt = new RedisLookupBolt(poolConfig, wordCountLookupMapper);
        PrintBolt printBolt = new PrintBolt();

        TopologyBuilder builder = new TopologyBuilder();
        builder.setSpout("wordSpout", wordSpout);
        builder.setBolt("redisLookupBolt", redisLookupBolt).shuffleGrouping("wordSpout");
        builder.setBolt("printBolt", printBolt).shuffleGrouping("redisLookupBolt");

        Config config = new Config();
        LocalCluster localCluster = new LocalCluster();
        localCluster.submitTopology("lookup-topology", config, builder.createTopology());
    }
}
